package com.lagou.client;

import com.lagou.service.UserService;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * 客户端端启动器
 */
@Component
public class ClientStarter implements ApplicationRunner {

    private static Map<String,UserService> servers = new HashMap<String, UserService>();

    /**
     * 上下文初始化完成后启动服务
     *
     * @param args
     * @throws Exception
     */
    @Override
    public void run(ApplicationArguments args) throws Exception {
        //创建zk客户端
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000,1);
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("192.168.3.110:2181")
                .sessionTimeoutMs(5000)
                .connectionTimeoutMs(3000)
                .retryPolicy(retryPolicy)
                .build();
        client.start();

        // 2、读取指定路径下所有node，并注册（缓存服务对象）服务提供者
        String path = "/lg-curator";

        // 判断根节点是否存在，不存在创建节点
        Stat stat = client.checkExists().forPath(path);
        if(stat == null){
            client.create().withMode(CreateMode.PERSISTENT).forPath(path,"".getBytes());
        }

        List<String> nodes = client.getChildren().forPath(path);
        if (nodes != null && nodes.size() > 0 ){
            for (String node : nodes) {
                addServer(client,path + "/" + node);
            }
        }


        // 3、对指定路径的节点进行监听，实现服务的动态发现与下线
        PathChildrenCache cache = new PathChildrenCache(client,path,true);
        PathChildrenCacheListener listener = new PathChildrenCacheListener() {
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                ChildData data = event.getData();
                switch (event.getType()) {
                    case CHILD_ADDED: //子节点增加
                        addServer(client,data.getPath());
                        break;
                    case CHILD_REMOVED: //子节点删除
                        String context = new String(data.getData());
                        removeServer(context);
                        break;
                    default:
                        break;
                }
            }
        };
        cache.getListenable().addListener(listener);
        cache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);


        // 创建临时节点，记录调用响应时间
        String recordPath = "/lg-record";
        //初始化节点调用信息，记录服务提供者和上次调用时间
        Set<String> set = servers.keySet();
        //使用简单的字符串进行数据记录 格式为 ip:端口::时间戳,ip:端口::时间戳
        String recordContext = "";
        for (String s : set) {
            //由于均为发生调用，故调用时间设置为0
            recordContext = recordContext+s+"::0,";
        }
        //创建临时节点，进行服务提供者的初始调用记录
        client.create().creatingParentsIfNeeded()
                .withMode(CreateMode.EPHEMERAL).forPath(recordPath, recordContext.getBytes());


        //4、循环给服务器写数据
        while (true){
            //节点信息初始化
            recordContext = "";

            Object[] serverArray = set.toArray();
            int length = serverArray.length;
            if (length > 0){
                //获取服务调用记录，调用最后一次响应请求时间小的进行调用，如果调用时间一样，随机选择一个调用
                //最后一次响应时间距当前时间超过5秒的，将时间清零
                Stat stat = new Stat();
                byte[] bytes = client.getData().storingStatIn(stat).forPath(recordPath);
                //获取请求记录数组
                String responseStr = new String(bytes);
                String[] requestRecords = responseStr.split(",");
                //调用服务地址
                String targetAddr = "";
                // 时间戳
                long lastTime = 0L;
                //遍历记录内容，进行判断哪个调用时间小
                for (int i = 0;i<requestRecords.length;i++){
                    String requestRecord = requestRecords[i];
                    //服务地址及最后一次请求的时间戳
                    String[] record = requestRecord.split("::");
                    long timeStamp = Long.valueOf(record[1]);
                    if (i == 0){
                        lastTime = Long.valueOf(record[1]);
                        targetAddr = record[0];
                    } else {
                        if (timeStamp <= lastTime){
                            lastTime = timeStamp;
                            targetAddr = record[0];
                        }
                    }
                }

                // 获取请求地址后，对请求记录进行重新计时设置
                for (String requestRecord : requestRecords) {
                    //服务地址及最后一次请求的时间戳
                    String[] record = requestRecord.split("::");
                    long timeStamp = Long.valueOf(record[1]);
                    //对于记录的时间戳大于0且距当前时间大于5秒的记录，进行清零操作
                    if (timeStamp > 0 && System.currentTimeMillis() - timeStamp > 5000){
                        recordContext = recordContext + record[0] + "::0,";
                    } else if (targetAddr.equals(record[0])){
                        //对于当前调用的请求，记录请求调用时间戳
                        recordContext = recordContext+record[0]+"::" + System.currentTimeMillis() + ",";
                    }else {
                        //未调用且未超时的不做处理
                        recordContext = recordContext + requestRecord + ",";
                    }
                }

                //更新记录节点的数据
                client.setData().withVersion(stat.getVersion()).forPath(recordPath,recordContext.getBytes());
                int index = (int) (Math.random()*length);
                UserService service = servers.get(targetAddr);
                String result = service.sayHello("are you ok !!");
                System.out.println(result);
            } else {
                System.out.println("没有服务提供者");
            }
            Thread.sleep(2000);
        }
    }

    /**
     * 缓存服务提供者
     * @param client
     * @param path
     * @throws Exception
     */
    public static void addServer(CuratorFramework client,String path) throws Exception {
        Stat stat = new Stat();
        byte[] bytes = client.getData().storingStatIn(stat).forPath(path);
        String server = new String(bytes);
        System.out.println("新服务提供者加入：" + server);
        String[] serverPath = new String(bytes).split(":");
        String ip = serverPath[0];
        int port = Integer.valueOf(serverPath[1]);
        RpcConsumer rpcConsumer = new RpcConsumer();
        UserService service = rpcConsumer.createProxy(UserService.class,ip,port);
        servers.put(server,service);
    }

    /**
     * 删除下线的服务
     * @param context
     */
    public static void removeServer(String context){
        System.out.println("服务提供者下线：" + context);
        servers.remove(context);
    }
}
